/*
 * Copyright (C) 2015 kmy-framework-base Project
 *               Author: Administrator
 *               Date: 2015年11月4日
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.zmq.ha.client.core;

import java.io.IOException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

import com.rabbitmq.client.AMQP.Queue;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Consumer;
import com.zmq.ha.client.declaration.Binding;
import com.zmq.ha.client.declaration.ConsumerDeclaration;
import com.zmq.ha.client.declaration.ConsumerDelegate;
import com.zmq.ha.client.declaration.QueueDeclaration;
import com.zmq.ha.client.declaration.ResourceDeclaration;
import com.zmq.ha.client.resource.RetryableSource;

/**
 * 
 * Name: ChannelWrapper.java
 * ProjectName: [kmy-framework-base]
 * Package: [com.kmy.framework.rabbitmq.recover.core.ChannelWrapper.java]
 * Description: Channel包装器,外部对于Channel的操作都会通过Channel包装器反射完成 ，包装器存储相关信息 
 * 
 * @since JDK1.7
 * @see
 *
 * Author: @author: Chris
 * Date: 2015年11月4日 下午4:47:34
 *
 * Update-User: @author
 * Update-Time:
 * Update-Remark:
 * 
 * Check-User:
 * Check-Time:
 * Check-Remark:
 * 
 * Company: kmy
 * Copyright: kmy
 */
public class ChannelWrapper extends RetryableSource implements InvocationHandler {

	//最近一次生成的queue名称
	public volatile String lastGeneratedQueueName;
	
	//上一次最大的DeliveryTag
	public volatile long previousMaxDeliveryTag;
	
	//最大的DeliveryTag
	public volatile long maxDeliveryTag;

	//map< consumer's name : consumerDeclaration >
	private final ConcurrentHashMap<String, ConsumerDeclaration> consumerDeclarations = new ConcurrentHashMap<String, ConsumerDeclaration>();

	public Channel delegate;

	public Channel proxy;

	private ConnectionWrapper connectionWrapper;
	
	public final ReentrantLock channelRecoveryLock = new ReentrantLock();
	
	private ResourceDeclaration basicQos;


	public ChannelWrapper(Channel source, ConnectionWrapper connectionWrapper) {
		delegate = source;
		this.connectionWrapper = connectionWrapper;
	}

	@Override
	public Object invoke(Object proxy, final Method method, final Object[] args) throws Throwable {
		deal(new Callable<Object>() {

			@Override
			public Object call() throws Exception {
				
				if( connectionWrapper.isRecovery){
					ConnectionWrapper.reentrantLock.lock();
					try{
						ConnectionWrapper.connectionRecoveryCondition.await();
					}finally{
						ConnectionWrapper.reentrantLock.unlock();
					}
				}
				
				String methodName = method.getName();
				boolean messageCheckHandle = "basicAck".equals(methodName) || "basicNack".equals(methodName)
						|| "basicReject".equals(methodName);
				
				if (messageCheckHandle) {
					
					long deliveryTag = (Long) args[0] - previousMaxDeliveryTag;
					
					if (deliveryTag > 0){
						
						args[0] = deliveryTag;
					}
					else{
						log.error("RMQ: error with deliverTag less then zero...");
						return null;
					}
					
				} else if ("basicConsume".equals(methodName))
					return handleConsumerDeclare(method, args);
				else if ("basicCancel".equals(methodName) && args[0] != null)
					consumerDeclarations.remove((String) args[0]);
				else if ("exchangeDelete".equals(methodName) && args[0] != null)
					connectionWrapper.exchangeDeclarations.remove((String) args[0]);
				else if ("exchangeUnbind".equals(methodName) && args[0] != null)
					connectionWrapper.exchangeBindings.remove((String) args[0], new Binding(args));
				else if ("queueDelete".equals(methodName) && args[0] != null)
					connectionWrapper.nameMapQueueDeclarations.remove((String) args[0]);
				else if ("queueUnbind".equals(methodName) && args[0] != null)
					connectionWrapper.queueBindings.remove((String) args[0], new Binding(args));
				
				Object result = method.invoke(delegate, args);

				if ("exchangeDeclare".equals(methodName))
					handleExchangeDeclare(method, args);
				else if ("exchangeBind".equals(methodName))
					handleExchangeBind(args);
				else if ("queueDeclare".equals(methodName))
					handleQueueDeclare(((Queue.DeclareOk) result).getQueue(), method, args);
				else if ("queueBind".equals(methodName))
					handleQueueBind(method, args);
				else if ("basicQos".equals(methodName)) {
			          if (args.length < 3 || !(Boolean) args[2])
			            basicQos = new ResourceDeclaration(method, args);
			        }

				return result;
			}
		}, false );

		return null;
	}
	
	
	 private void migrateConfiguration(Channel channel) throws Exception {
	    channel.setDefaultConsumer(delegate.getDefaultConsumer());
	    if (basicQos != null)
	      basicQos.invoke(channel);
	  }

	// 已经持久话的无需，重启，mq会保存
	void handleExchangeDeclare(Method method, Object[] args) {
		log.debug("Declaring Exchange [ Exchange : "+ (String) args[0]+" ]");
		boolean autoDelete = args.length > 3 && (Boolean) args[3];
		boolean durable = args.length > 2 && (Boolean) args[2];
		if (autoDelete || !durable)
			connectionWrapper.nameMapExchangeDeclaration.put((String) args[0], new ResourceDeclaration(method, args));
	}

	private String handleConsumerDeclare(Method method, Object[] args) throws Exception {
		String queueName = "".equals(args[0]) ? lastGeneratedQueueName : (String) args[0];
		log.debug("Declaring Consumer [ Consumer : "+queueName+" ]");
		Consumer consumer = (Consumer) args[args.length - 1];
		args[args.length - 1] = new ConsumerDelegate(this, consumer);
		String consumerTag = (String) method.invoke(delegate, args);
		QueueDeclaration queueDeclaration = connectionWrapper.nameMapQueueDeclarations.get(queueName);
		if (queueDeclaration != null)
			queueName = queueDeclaration.queueName;

		consumerDeclarations.put(consumerTag, new ConsumerDeclaration(method, args, queueDeclaration));
		
		return consumerTag;
	}

	private void handleExchangeBind(Object[] args) {
		log.debug("binding exchange [ exchange : "+(String) args[0] +" ]");
		ArrayList<Binding> bindings = connectionWrapper.exchangeBindings.get((String) args[0]);
		if (bindings == null) {
			bindings = new ArrayList<>();
			connectionWrapper.exchangeBindings.put((String) args[0], bindings);
		}
		bindings.add(new Binding(args));

	}

	private void handleQueueBind(Method method, Object[] args) {
		log.debug("binding Queue [ Queue : "+(String) args[0] +" ]");
		ArrayList<Binding> bindings = connectionWrapper.queueBindings.get((String) args[0]);
		if (bindings == null) {
			bindings = new ArrayList<>();
			connectionWrapper.queueBindings.put((String) args[0], bindings);
		}
		bindings.add(new Binding(args));

	}

	private void handleQueueDeclare(String queueName, Method method, Object[] args) {
		log.debug("Declaring Queue [ Queue : "+queueName +" ]");

		if (args == null)
			lastGeneratedQueueName = queueName;

		boolean autoDelete = args == null || (Boolean) args[3];
		boolean durable = args != null && (Boolean) args[1];
		if (autoDelete || !durable)
			connectionWrapper.nameMapQueueDeclarations.put(queueName, new QueueDeclaration(method, args, queueName));
	}

	
	public synchronized void recoverChannel(boolean viaConnectionRecovery) throws Exception {

		try {
			delegate = deal(new Callable<Channel>() {
				@Override
				public Channel call() throws Exception {
					log.info("Recovering {}", ChannelWrapper.this);
					previousMaxDeliveryTag = maxDeliveryTag;
					Channel channel = connectionWrapper.createChannel(delegate.getChannelNumber());
					migrateConfiguration(channel);
					return channel;
				}
			}, true);
			recoverConsumers(!viaConnectionRecovery);
			
		} catch (Exception e) {
		} finally {
			if( !viaConnectionRecovery ){
				connectionWrapper.isRecovery = false;
				try{
					ConnectionWrapper.connectionRecoveryCondition.signalAll();
				}finally{
					ConnectionWrapper.reentrantLock.unlock();
				}
			}
				
		}
	}

	
	private void recoverConsumers(boolean recoverReferences) throws Exception {
		log.debug("building recoverConsumers");
		Set<QueueDeclaration> recoveredQueues = new HashSet<QueueDeclaration>();
		Set<String> recoveredExchanges = new HashSet<String>();

		for (Iterator<Map.Entry<String, ConsumerDeclaration>> it = consumerDeclarations.entrySet().iterator(); it
				.hasNext();) {
			Map.Entry<String, ConsumerDeclaration> entry = it.next();
			ConsumerDeclaration consumerDeclaration = entry.getValue();
			Object[] args = consumerDeclaration.args;
			ConsumerDelegate consumer = (ConsumerDelegate) args[args.length - 1];
			String queueName = consumerDeclaration.queueDeclaration != null
					? consumerDeclaration.queueDeclaration.queueName : (String) args[0];

			try {
				
				if (recoverReferences) {
					ArrayList<Binding> queueBindings = connectionWrapper.queueBindings.get(queueName);
					recoverRelatedExchanges(recoveredExchanges, queueBindings);
					if (consumerDeclaration.queueDeclaration != null
							&& recoveredQueues.add(consumerDeclaration.queueDeclaration))
						queueName = recoverQueue(queueName, consumerDeclaration.queueDeclaration, queueBindings);
				}
				//主动停时，让系统消耗掉线程残余的消息
				Thread.sleep(2000);
				
				consumer.open();
				consumerDeclaration.invoke(delegate);
				
			} catch (Exception e) {
				throw e;
			}
		}
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	private void recoverRelatedExchanges(Set<String> recoveredExchanges, List<Binding> queueBindings) throws Exception {
		synchronized (queueBindings) {
			log.debug("building recoverRelatedExchanges");
			for (Binding queueBinding : queueBindings) {
				String exchangeName = queueBinding.source;
				if (recoveredExchanges.add(exchangeName)) {
					ResourceDeclaration exchangeDeclaration = connectionWrapper.exchangeDeclarations.get(exchangeName);
					if (exchangeDeclaration != null)
						recoverExchange(exchangeName, exchangeDeclaration);

					HashMap map = new HashMap();
					map.put(null, connectionWrapper.exchangeBindings.get(exchangeName));
					recoverExchangeBindings(map.values().iterator());
					map = null;// TODO bad code  (>....<)

				}
			}
		}
	}

	/** Recovers the {@code queueName} along with its {@code queueBindings}. */
	@SuppressWarnings({ "unused", "rawtypes", "unchecked" })
	private String recoverQueue(String queueName, QueueDeclaration queueDeclaration, ArrayList<Binding> queueBindings)
			throws Exception {
		log.debug("building recoverQueue");
		String newQueueName = queueName;

		if (queueDeclaration != null) {
			newQueueName = recoverQueue(queueName, queueDeclaration);

			// Update dependencies for new queue names
			if (!queueName.equals(newQueueName)) {
				connectionWrapper.nameMapQueueDeclarations.remove(queueName);
				connectionWrapper.nameMapQueueDeclarations.put(newQueueName, queueDeclaration);
				connectionWrapper.updateQueueBindingReferences(queueName, newQueueName);
			}
		}
		HashMap map = new HashMap();
		map.put(null, queueBindings);
		recoverQueueBindings(map.values().iterator());

		return newQueueName;
	}

	@Override
	public Channel getRecoveryChannel() throws IOException {
		return delegate;
	}

	@Override
	public boolean throwOnRecoveryFailure() {
		// TODO Auto-generated method stub
		return false;
	}

	public ConnectionWrapper getConnectionWrapper() {
		return connectionWrapper;
	}
	
	
	public void channelShutdown() {
	    synchronized (consumerDeclarations) {
	      for (ResourceDeclaration invocation : consumerDeclarations.values())
	    		 ((ConsumerDelegate) invocation.args[invocation.args.length - 1]).close();
	    }
	  }
	
}
